package com.wlz.demo.consumer;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

/**
 * @author wlz
 * @date 2020-10-29
 *   rocketmq 消费端
 */
public class SimpleConsumer {


    public static void main(String[] args) throws InterruptedException, MQClientException {
        // 创建一个消息消费者，并设置一个消息消费者组
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("groupTest");

        // 指定 NameServer 地址
        consumer.setNamesrvAddr("192.168.27.133:9876");

        // 设置 Consumer 第一次启动时从队列头部开始消费还是队列尾部开始消费
//        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

        // 订阅指定 Topic 下的所有消息
        consumer.subscribe("topicTest", "*");
        // Register callback to execute on arrival of messages fetched from brokers.
        //注册消息监听器
        //单线程对应单队列
//        consumer.registerMessageListener(new MessageListenerOrderly() {
////            @Override
////            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
////                return null;
////            }
//            @Override
//            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs,
//                                                       ConsumeOrderlyContext context) {
//                for (MessageExt msg : msgs) {
//                    System.out.printf("%s Receive New Messages: %s %s 延迟时间 %s %n", Thread.currentThread().getName(), new String(msg.getBody()),msg.getQueueId(),(System.currentTimeMillis()-msg.getStoreTimestamp()));
//                }
//                return ConsumeOrderlyStatus.SUCCESS;
//            }
//        });

        //注册消息监听器
        //Concurrently多线程无序消费
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                            ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
                    System.out.printf("%s Receive New Messages: %s %s %n", Thread.currentThread().getName(), new String(msg.getBody()),msg.getQueueId());
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        // 消费者对象在使用之前必须要调用 start 初始化
        consumer.start();
        System.out.printf("Consumer Started.%n");
    }
}
